package ex.datastream.functions.sourceFunction;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 *  
 */
public class SourceFuncByItem implements SourceFunction<SourceFuncByItem.Item> {
    private boolean isRunning = true;

    @Override
    public void run(SourceContext<Item> ctx) throws Exception {
        while (isRunning) {
            Item item = generateItem();
            ctx.collect(item);
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {

    }

    private Item generateItem() {
        int i = new Random().nextInt(100);
        Item item = new Item();
        item.setName("name" + i);
        item.setId(i);
        return item;
    }

    public class Item {
        private String name;
        private Integer id;

        @Override
        public String toString() {
            return "Item{" +
                    "name='" + name + '\'' +
                    ", id=" + id +
                    '}';
        }

        public Item() {
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public Integer getId() {
            return id;
        }

        public void setId(Integer id) {
            this.id = id;
        }
    }

}
